package com.atguigu.gmall.realtime.app.dwd;

import com.atguigu.gmall.realtime.util.MySqlUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class DwdTradeCartAddApp {


    // flinksql
    // 0 环境
    // 1  数据源
    // 1.1  事实表  topic_db ->  cart_info   kafka
    // 1.2  维度表 base_dic    mysql
    // 2  lookup_join  sql
    // 3  目标表  dwd_trade_cart_add
    // 4  写入目标表
    public static void main(String[] args) {
        // 0 环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        // 1  数据源
        // 1.1  事实表  topic_db ->  cart_info   kafka
           String createTopicDbSQL="     CREATE TABLE topic_db (\n" +
                       "        `database` STRING,\n" +
                        "       `table` STRING,\n" +
                        "        `type` STRING,\n" +
                        "         `data` MAP<STRING,STRING>,\n" +
                       "          `old` MAP<STRING,STRING> ,\n" +
                       "          `proc_time`  as proctime(), \n"+
                        "         ts   STRING \n"+
                        "          ) WITH (\n" +
                         "          'connector' = 'kafka',\n" +
                        "          'topic' = 'topic_db',\n" +
                        "          'properties.bootstrap.servers' = 'hadoop102:9092',\n" +
                        "          'properties.group.id' = 'dwd_trade_cart_add_app',\n" +
                       "          'scan.startup.mode' = 'group-offsets',\n" +
                        "           'format' = 'json'\n" +
                        "         )";

           tableEnv.executeSql(createTopicDbSQL);
           // select * from
        //   topic_db where table='cart_info'
          //  and  (type= 'insert'  or  (type= 'update'  and    (cast  data['sku_num']  as INTEGER )-(cast  old['sku_num']  as INTEGER ) >0) )
           String selectCartAddSQL="select \n" +
                   "`data`['id'] id,\n" +
                   "`data`['user_id'] user_id,\n" +
                   "`data`['sku_id'] sku_id,\n" +
                   "`data`['cart_price'] cart_price,\n" +
                   " if(type= 'insert',`data`['sku_num'],(cast (  (cast( `data`['sku_num']  as INTEGER)  - cast(`old`['sku_num']  as INTEGER))   as string))   ) sku_num,\n" +
                   "`data`['sku_name'] sku_name,\n" +
                   "`data`['is_checked'] is_checked,\n" +
                   "`data`['create_time'] create_time,\n" +
                   "`data`['operate_time'] operate_time,\n" +
                   "`data`['is_ordered'] is_ordered,\n" +
                   "`data`['order_time'] order_time,\n" +
                   "`data`['source_type'] source_type, \n" +
                   " proc_time ,ts \n" +
                   " from\n" +
                   "         topic_db where `table`='cart_info'\n" +
                   "            and  (type= 'insert'  or " +
                   " (type= 'update'  and    (cast( `data`['sku_num']  as INTEGER) )-(cast(`old`['sku_num']  as INTEGER) ) >0) )\n"  ;


        Table cartAddTable = tableEnv.sqlQuery(selectCartAddSQL);

        tableEnv.createTemporaryView("cart_add_info",cartAddTable);
     //   tableEnv.executeSql("select * from "+cartAddTable).print();

        // 1.2  维度表 base_dic    mysql
        String createBaseDicSQL="   CREATE   TABLE base_dic (\n" +
                "               dic_code STRING,\n" +
                "               dic_name STRING,\n" +
                "              PRIMARY KEY (dic_code) NOT ENFORCED\n" +
                "            ) WITH (\n" +
                "              'connector' = 'jdbc',\n" +
                 "             'url' = 'jdbc:phoenix:hadoop102:2181',\n"+
                "              'table-name' = 'GMALL2022.DIM_BASE_DIC',\n"+
                "               'lookup.cache.max-rows' ='1000',\n" +
                "               'lookup.cache.ttl' ='60s'\n" +
                "             ) ";

      //  tableEnv.executeSql(createBaseDicSQL);

        //为了节省教学开销
        String baseDicDDL = MySqlUtil.getBaseDicDDL();
        tableEnv.executeSql(baseDicDDL);

        // 2  lookup_join  sql
        String lookupJoinSQL="select  ci.*, dic_name as source_type_name  from  cart_add_info ci \n" +
                " join  base_dic FOR SYSTEM_TIME AS OF ci.proc_time AS dic\n" +
                " on  ci.source_type=dic.dic_code";
        Table lookupJoinTable = tableEnv.sqlQuery(lookupJoinSQL);

     //   tableEnv.executeSql("select * from "+ lookupJoinTable).print();

        // 3  目标表  dwd_trade_cart_add

        tableEnv.executeSql("" +
                        "create table dwd_trade_cart_add(\n" +
                        "id string,\n" +
                        "user_id string,\n" +
                        "sku_id string,\n" +
                        "source_type_code string,\n" +
                        "source_type_name string,\n" +
                        "sku_num string,\n" +
                        "ts string,\n" +
                        "PRIMARY KEY (id) NOT ENFORCED \n"  +
                        ") with(   " +
                        "    'connector' = 'upsert-kafka', \n" +
                        "                      'topic' = 'dwd_trade_cart_add', \n" +
                        "                      'properties.bootstrap.servers' = 'hadoop102:9092', \n" +
                        "                          'key.format' = 'json' , \n" +
                        "                        'value.format' = 'json'  ) "  );

        // 4  写入目标表
        tableEnv.executeSql("" +
                "insert into dwd_trade_cart_add select id,user_id,sku_id," +
                "source_type,source_type_name,sku_num,ts  from "+lookupJoinTable);


    }
}
